package com.hcy.mqbusiness.mqcenterfororder;

import com.hcy.mqbusiness.MqAsyncProducer;
import com.hcy.mqbusiness.MqProducer;
import com.hcy.rabbitbasicmodel.model.ResponseData;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

/**
 * @Author huchenying
 * @Description
 * @Date 2022/3/15 14:40
 */
@Service
public class RocketMqAsyncProducer implements MqAsyncProducer {

    private final DefaultMQProducer asyncProducer = new DefaultMQProducer("asyncProducer");

    static Logger logger = LoggerFactory.getLogger(RocketMqAsyncProducer.class);

    @Value("${MQ.namesrvAddr}")
    private String namesrvAddr;

    @PostConstruct
    public void init(){
        asyncProducer.setNamesrvAddr(namesrvAddr);
        // 设置配置，可以从配置中拿，暂时使用固定数据
        //异步失败重试
        asyncProducer.setRetryTimesWhenSendAsyncFailed(2);
        //设置新创建的Topic的Queue数量为1
        asyncProducer.setDefaultTopicQueueNums(1);
        asyncProducer.setSendMsgTimeout(3000);
        //设置vip通道，防止消费超时
        asyncProducer.setSendMessageWithVIPChannel(false);
        try {
            asyncProducer.start();
        } catch (MQClientException e) {
            logger.error("syncProducer 启动失败！");
            e.printStackTrace();
        }
    }

    /**
     * @Author huchenying
     * @Description 发送一个同步消息
     * @Date 2022/3/15 14:45
     **/
    public void sendSyncMessage(Message message){
        try {
            asyncProducer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    logger.info("异步消息发送成功 -> {}" , ToStringBuilder.reflectionToString(sendResult) );
                }

                @Override
                public void onException(Throwable throwable) {
                    logger.error("异步消息发送失败 -> {}" , ToStringBuilder.reflectionToString(throwable) );
                }
            });
        } catch (MQClientException | RemotingException | InterruptedException e) {
            logger.error("服务器异常，异步消息发送失败！");
        }
    }
    /**
     * @Author huchenying
     * @Description 发送一个同步消息
     * @Date 2022/3/15 14:45
     **/
    public void sendSyncMessageList(List<Message> messageList){
        try {
            asyncProducer.send(messageList, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    logger.info("异步消息发送成功 -> {}" , ToStringBuilder.reflectionToString(sendResult) );
                }
                @Override
                public void onException(Throwable throwable) {
                    logger.error("异步消息发送失败 -> {}" , ToStringBuilder.reflectionToString(throwable) );
                }
            });
        } catch (MQClientException | RemotingException | InterruptedException | MQBrokerException e) {
            logger.error("服务器异常，异步消息发送失败！");
        }
    }

    @PreDestroy
    public void shudown(){
        asyncProducer.shutdown();
    }

}
